/**
 * 
 */
package edu.umd.clip.lm.util;

import java.util.*;
import java.util.concurrent.locks.*;

import edu.berkeley.nlp.util.Pair;
/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class DataSequencer<T> {
	private PriorityQueue<Pair<Integer, T>> queue;
	private Lock lock;
	private Condition hasItemsCondition;
	private int nextItem = 1;
	private int lastItem = nextItem - 2;
	private boolean lastItemSet = false;
	
	private class Comp implements Comparator<Pair<Integer, T>> {
		public int compare(Pair<Integer, T> arg0, Pair<Integer, T> arg1) {
			return arg0.getFirst().compareTo(arg1.getFirst());
		}
	}
	
	public DataSequencer() {
		queue = new PriorityQueue<Pair<Integer, T>>(20, new Comp());
		lock = new ReentrantLock();
		hasItemsCondition = lock.newCondition();
	}

	public DataSequencer(int firstItem) {
		this();
		nextItem = firstItem;
		lastItem = nextItem - 2;
	}
	
	public void putItem(Pair<Integer, T> pair) {
		lock.lock();
		queue.add(pair);
		Pair<Integer, T> first = queue.peek();
		if (first.getFirst().intValue() == nextItem) {
			hasItemsCondition.signalAll();
		}
		lock.unlock();
	}
	
	public Pair<Integer, T> getFirst() throws InterruptedException {
		lock.lock();
		Pair<Integer, T> first = queue.peek();
		while(!_isDone() && (first == null || first.getFirst().intValue() > nextItem)) {
			try {
				hasItemsCondition.await();
				first = queue.peek();
			} catch(InterruptedException e) {
				lock.unlock();
				throw e;
			}
		}
		if (first != null) {
			first = queue.remove();
			++nextItem;
		}
		lock.unlock();
		return first;
	}
	
	public Pair<Integer, T> getFirstNonBlock() {
		lock.lock();
		Pair<Integer, T> first = queue.peek();
		if (first.getFirst().intValue() == nextItem) {
			first = queue.remove();
			++nextItem;
		} else {
			first = null;
		}
		lock.unlock();
		return first;
	}

	public void setLastItem(int lastItem) {
		lock.lock();
		this.lastItem = lastItem;
		this.lastItemSet = true;
		hasItemsCondition.signalAll();
		lock.unlock();
	}
	
	private boolean _isDone() {
		return lastItemSet && (lastItem == nextItem - 1);
	}
	
	public boolean isDone() {
		lock.lock();
		boolean result = _isDone();
		lock.unlock();
		return result;
	}
}
